package com.bobo.gmall.realtime.app.function;

import com.alibaba.fastjson.JSONObject;
import com.bobo.gmall.realtime.common.GmallConfig;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Collection;
import java.util.Collections;
import java.util.Set;

public class MyJdbcSink extends RichSinkFunction<JSONObject> {

    Connection conn;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName(GmallConfig.MYSQL_DRIVER);
        conn = DriverManager.getConnection(GmallConfig.MYSQL_SERVER);
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
//        JSONObject before = value.getJSONObject("before");
        JSONObject after = value.getJSONObject("after");
        String type = value.getString("type");
        String db = value.getString("database");
        String tableName = value.getString("tableName");
        String sinkTable = value.getString("sinkTable");

        Set<String> keys = after.keySet();
        Collection<Object> values = after.values();

        String upsertSql = genUpsertSql(sinkTable, keys, values);
        System.out.println(upsertSql);

        PreparedStatement preparedStatement = conn.prepareStatement(upsertSql);
        preparedStatement.execute();
    }

//    replace into gmall_flink.base_trademark (id,tm_name,logo_url)
//    values (13,"ttt","ttt")
    private String genUpsertSql(String sinkTable, Set<String> keys, Collection<Object> values){
        return "replace into "+GmallConfig.MYSQL_SCHEMA+"."+sinkTable+
                " ("+ StringUtils.join(keys,",")+")"+
                " values ('"+StringUtils.join(values,"','")+"')";
    }
}
